package com.example;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.util.concurrent.atomic.AtomicInteger;

public class Main {
    private static final Logger log = LoggerFactory.getLogger(Main.class);

    public static void main(String[] args) {
        log.info("======== TEST TOPIC DESTINATION ========");
        Destination topicDestination = new ActiveMQTopic("test.topic");
        testDestination(topicDestination);

        log.info("======== TEST QUEUE DESTINATION ========");
        Destination queueDestination = new ActiveMQQueue("test.queue");
        testDestination(queueDestination);
    }

    public static MessageListener createMessageListener(int consumerId, Destination destination) {
        final AtomicInteger count = new AtomicInteger();
        return message -> {
            try {
                String msg;
                if (message instanceof TextMessage) {
                    msg = ((TextMessage) message).getText();
                } else {
                    msg = message.toString();
                }
                message.acknowledge();
                log.info("[{}] {} => receive from {}: {}.", consumerId, count.incrementAndGet(), destination.toString(), msg);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        };
    }

    public static void testDestination(Destination destination) {
        try {
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
            ActiveMQConnection conn = (ActiveMQConnection) factory.createConnection();
            conn.start();

            Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);

            // 创建消费者
            session.createConsumer(destination).setMessageListener(createMessageListener(0, destination));
            session.createConsumer(destination).setMessageListener(createMessageListener(1, destination));

            // 创建生产者
            MessageProducer producer = session.createProducer(destination);
            int index = 0;
            while (index++ < 10) {
                TextMessage message = session.createTextMessage(index + " message.");
                producer.send(message);
            }

            Thread.sleep(5000);
            session.close();
            conn.close();

        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
